package org.wheel.plugins.toolkit.sequence;

import lombok.RequiredArgsConstructor;
import org.wheel.plugins.toolkit.cache.core.SequenceProperties;
import org.wheel.plugins.toolkit.cache.spi.CacheService;
import org.wheel.plugins.toolkit.cache.spi.LockProvider;
import org.wheel.plugins.toolkit.cache.spi.SequenceAllocator;

import java.util.concurrent.atomic.AtomicLong;

@RequiredArgsConstructor
public class CacheSequenceAllocator implements SequenceAllocator {

    private final CacheService cacheService;
    private final LockProvider lockProvider;
    private final SequenceProperties props;

    private final java.util.Map<String, AtomicLong> localCache = new java.util.concurrent.ConcurrentHashMap<>();
    private final java.util.Map<String, Long> maxCache = new java.util.concurrent.ConcurrentHashMap<>();

    @Override
    public long nextId(String key) {
        String fullKey = "seq:" + key;
        AtomicLong current = localCache.computeIfAbsent(fullKey, k -> new AtomicLong(loadRange(k)));
        long next = current.incrementAndGet();
        if (next >= maxCache.get(fullKey)) {
            synchronized (this) {
                if (next >= maxCache.get(fullKey)) {
                    loadRange(fullKey);
                    next = current.incrementAndGet();
                }
            }
        }
        return next;
    }

    private long loadRange(String fullKey) {
        String lockKey = fullKey + ":lock";
        if (!lockProvider.tryLock(lockKey, 60)) {
            throw new RuntimeException("Unable to acquire sequence lock for: " + fullKey);
        }
        try {
            String val = cacheService.getValue(fullKey);
            long current = val == null ? 0 : Long.parseLong(val);
            long nextMax = current + props.getCacheLength();
            cacheService.setKey(fullKey, String.valueOf(nextMax));
            maxCache.put(fullKey, nextMax);
            return current;
        } finally {
            lockProvider.unlock(lockKey);
        }
    }
}
